package com.lxl.mayi;

import org.apache.commons.collections.CollectionUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @Author lixilin
 * @Date 2022/1/1 11:18
 * @Description
 */
public class UserCarbonReducer extends Reducer<UserCarbon, UserCarbon, UserCarbon, NullWritable> {
    private static final Logger LOG = LoggerFactory.getLogger(UserCarbonReducer.class);
    NullWritable nullWritable = NullWritable.get();
    private static int DATE_DIFF_ONE =1;
    SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy/mm/dd");

    UserCarbon out = new UserCarbon();
    @Override
    protected void reduce(UserCarbon key, Iterable<UserCarbon> values, Context context) throws IOException, InterruptedException {

        List<UserCarbon> tmpList = new ArrayList<>();
        for (UserCarbon value : values) {
            // 1,判断list是否为空,为空直接添加进去
            if (CollectionUtils.isEmpty(tmpList)) {
                out.setUserId(value.getUserId());
                out.setDate(value.getDate());
                out.setCarbon(value.getCarbon());
                tmpList.add(out);
                continue;
            }
            // 2,判断当前value的时间和list中末尾的usercarbon的时间差是否为1
            String date = value.getDate();
            UserCarbon listLastUserCarbon = tmpList.get((tmpList.size() - 1));
            int dateDiff = getDateDiff(listLastUserCarbon.getDate(),date, dateFormat);
            if (dateDiff == DATE_DIFF_ONE) {
                // 时间是连续的直接添加进list
                out.setUserId(value.getUserId());
                out.setDate(value.getDate());
                out.setCarbon(value.getCarbon());
                tmpList.add(out);
                continue;
            }
            // 3,判断tmpList中的元素个数是否大于等于3
            if (tmpList.size() >=3) {
                // 将tmpList中之前连续时间超过3天的数据输出
                for (UserCarbon userCarbon : tmpList) {
                    context.write(userCarbon,nullWritable);
                }
            }
            // 4,清空tmpList,并把当前的value添加进去
            tmpList.clear();

            out.setUserId(value.getUserId());
            out.setDate(value.getDate());
            out.setCarbon(value.getCarbon());
            tmpList.add(out);
        }
        // 循环完成后,将tmpList中超过3天连续的数据进行输出
        if (tmpList.size() >=3) {
            // 将tmpList中之前连续时间超过3天的数据输出
            for (UserCarbon userCarbon : tmpList) {
                context.write(userCarbon,nullWritable);
            }
        }
    }

    private int getDateDiff(String dateStr1, String dateStr2,SimpleDateFormat dateFormat) {
        try {

            Date date1 = dateFormat.parse(dateStr1);
            Date date2 = dateFormat.parse(dateStr2);
            long days=(date1.getTime()-date2.getTime())/(1000*60*60*24);
            return (int) days;
        } catch (Exception e) {
            LOG.error("解析时间错误: ",e);
        }
        return Integer.MAX_VALUE;
    }
}
